package com.ada.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * Window Operations可以设置窗口的大小和滑动窗口的间隔来动态的获取当前Steaming的允许状态。基于窗口的操作会在一个比 StreamingContext 的批次间隔更长的时间范围内，通过整合多个批次的结果，计算出整个窗口的结果。
  */
object SparkStreaming07_Window {

    def main(args: Array[String]): Unit = {

        //创建SparkConf并初始化SSC
        val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming07_Window")
        val ssc = new StreamingContext(sparkConf, Seconds(3))

        //kafka topic
        val topic = "source"

        //通过KafkaUtil创建kafkaDSteam
        val kafkaDSteam: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream(
            ssc,
            "hadoop121:2181",
            topic,
            Map(topic -> 3),
            StorageLevel.MEMORY_ONLY
        )

        //窗口的大小应该为采集周期的整数倍，窗口滑动的步长也应该为采集周期的整数倍
        val windowStream: DStream[(String, String)] = kafkaDSteam.window(Seconds(9), Seconds(3))

        //将采集的数据进行分解（扁平化）
        val wordStream: DStream[String] = windowStream.flatMap(t => t._2.split(" "))

        //将数据进行结构的转换，方便统计分析
        val mapStream: DStream[(String, Int)] = wordStream.map((_, 1))

        //将转换后的数据进行聚合处理
        //这个操作不会进行累加（无状态更新）
        val wordToSumSteam: DStream[(String, Int)] = mapStream.reduceByKey(_ + _)

        //打印结果
        wordToSumSteam.print()

        //启动采集器
        ssc.start()
        //Drvier等待采集器的执行
        ssc.awaitTermination()
    }

}

